package trouger.dmt;

import java.util.ArrayList;

/**
 * Do default join operation between two source tables.
 * Do aggregation operations when AggregationOperator is specified.
 * Alter among inner/(left/right/full outer) joins by parameter.
 */
class JoinTableRecordGroupFilter
implements RecordGroupFilter{
	
	protected JoinType joinType;
	protected RecordOutputer recordOutputer;
	protected ArrayList<String[]> firstSourceRecords = new ArrayList<String[]>();
	protected int sourceIndex = -1;
	protected FieldIndexPair[] resultFieldsIndex;
	protected AggregationOperation[] aggregationOperation;
	protected AggregationOperator[] aggregationOperator;
	protected boolean firstSourceAggregation = false;
	protected boolean secondSourceAggregation = false;
	protected String[] secondSourceRecordForAggregation = null;
	protected boolean firstSourceOuterJoined;
	protected boolean secondSourceOuterJoined;
	protected boolean hasFirstSourceRecords = false;
	protected boolean hasSecondSourceRecords = false;
	protected boolean dontSetHasSecondSourceRecords = false;
	
	public void setForcedUniqueness(boolean firstSource, boolean secondSource){
		if (firstSource) this.firstSourceAggregation = true;
		if (secondSource) this.secondSourceAggregation = true;
		if (firstSource || secondSource){
			if (this.aggregationOperation == null){
				this.aggregationOperation = new AggregationOperation[0];
			}
		}
	}
	public void setJoinType(JoinType joinType){
		this.joinType = joinType;
		switch (joinType){
		case InnerJoin:
			firstSourceOuterJoined = secondSourceOuterJoined = false;
			break;
		case LeftOuterJoin:
			firstSourceOuterJoined = true;
			secondSourceOuterJoined = false;
			break;
		case RightOuterJoin:
			firstSourceOuterJoined = false;
			secondSourceOuterJoined = true;
			break;
		case FullOuterJoin:
			firstSourceOuterJoined = secondSourceOuterJoined = true;
		}
	}
	public void setResultFieldsIndex(FieldIndexPair[] resultFieldsIndex){
		this.resultFieldsIndex = resultFieldsIndex;
	}
	public void setAggregationOperations(AggregationOperation[] aggregationOperation){
		this.aggregationOperation = aggregationOperation;
		if (aggregationOperation != null){
			int n = aggregationOperation.length;
			aggregationOperator = new AggregationOperator[n];
			for (int i = 0; i < n; i++){
				try {
					aggregationOperator[i] = aggregationOperation[i].getOperatorClass().newInstance();
				} catch (Exception e) {
					throw new IllegalArgumentException("on creating AggregationOperator instance, " + e.toString());
				} 
				int srcIdx = aggregationOperation[i].getSourceField().getSourceIndex();
				if (srcIdx == 1) firstSourceAggregation = true;
				else if (srcIdx == 2) secondSourceAggregation = true;
				else {
					throw new IllegalStateException("default join with more than two source tables is not implemented");
				}
			}
		}
	}
	
	@Override
	public void setRecordOutputer(RecordOutputer recordOutputer) {
		this.recordOutputer = recordOutputer;
	}
	
	@Override
	public boolean beginRecordSource(int sourceIndex) {
		this.sourceIndex = sourceIndex;
		if (sourceIndex == 2 && firstSourceRecords.size() == 0){
			if (secondSourceOuterJoined){
				int n = 1;
				for (FieldIndexPair field : resultFieldsIndex){
					if (field.getSourceIndex() == 1){
						if (field.getColumnIndex() + 1 > n){
							n = field.getColumnIndex() + 1;
						}
					}
				}
				firstSourceRecords.add(new String[n]);
			}else{
				return false;
			}
		}
		return true;
	}

	@Override
	public boolean endRecordSource() {
		if (sourceIndex == 1){
			if (firstSourceRecords.size() == 0){
				return secondSourceOuterJoined;
			}
		}else if (sourceIndex == 2 && secondSourceAggregation){
			if (secondSourceRecordForAggregation == null){
				return firstSourceOuterJoined;
			}else{
				secondSourceAggregation = false;
				processRecord(secondSourceRecordForAggregation);
			}
		}
		return true;
	}
	
	@Override
	public void endOfGroup() {
		if (!hasSecondSourceRecords && firstSourceOuterJoined){
			this.sourceIndex = 2;
			this.secondSourceAggregation = false;
			int n = 1;
			for (FieldIndexPair field : resultFieldsIndex){
				if (field.getSourceIndex() == 2){
					if (field.getColumnIndex() + 1 > n){
						n = field.getColumnIndex() + 1;
					}
				}
			}
			this.dontSetHasSecondSourceRecords = true;
			this.processRecord(new String[n]);
		}
	}

	@Override
	public boolean processRecord(String[] record) {
		if (sourceIndex == 1){
			this.hasFirstSourceRecords = true;
			if (!firstSourceAggregation){
				int n;
				String[] record_copy = new String[n = record.length];
				for (int i = 0; i < n; i++){
					record_copy[i] = record[i];
				}
				firstSourceRecords.add(record_copy);
			}else{
				int n = aggregationOperation.length;
				int m = record.length;
				for (int i = 0; i < n; i++){
					FieldIndexPair field = aggregationOperation[i].getSourceField();
					if (field.getSourceIndex() == 1){
						int k = field.getColumnIndex();
						this.aggregationAcceptValue(i, k < m ? record[k] : null);
					}
				}
				if (firstSourceRecords.size() == 0){
					firstSourceRecords.add(record);
				}else{
					firstSourceRecords.set(0, record);
				}
			}
		}else if (sourceIndex == 2){
			if (!this.dontSetHasSecondSourceRecords){
				this.hasSecondSourceRecords = true;
			}
			if (!secondSourceAggregation){
				for (String[] record1 : firstSourceRecords){
					int n = resultFieldsIndex.length;
					Object[] resultRecord = new Object[n + 1];
					for (int i = 0; i < n; i++){
						int j = resultFieldsIndex[i].getSourceIndex();
						int k = resultFieldsIndex[i].getColumnIndex();
						resultRecord[i + 1] = j == 1 ? record1[k] : record[k];
					}
					if (aggregationOperation != null){
						int i = 0;
						for (AggregationOperation op : aggregationOperation){
							if (op.getSourceField().getSourceIndex() == 1 && hasFirstSourceRecords ||
							op.getSourceField().getSourceIndex() == 2 && hasSecondSourceRecords){
								resultRecord[op.getResultField()] = aggregationOperator[i].getAggregatedValue();
							}
							i++;
						}
					}
					recordOutputer.output(resultRecord);
				}
			}else{
				secondSourceRecordForAggregation = record;
				int n = aggregationOperation.length;
				int m = record.length;
				for (int i = 0; i < n; i++){
					FieldIndexPair field = aggregationOperation[i].getSourceField();
					if (field.getSourceIndex() == 2){
						int k = field.getColumnIndex();
						this.aggregationAcceptValue(i, k < m ? record[k] : null);
					}
				}
			}
		}else{
			throw new IllegalStateException("default join with more than two source tables is not implemented");
		}
		return true;
	}
	
	protected void aggregationAcceptValue(int aggIndex, String value){
		try{
			aggregationOperator[aggIndex].acceptValue(value);
		}catch(RuntimeException e){
			if (aggregationOperation[aggIndex].isResumeOnError()){
				System.err.println("Aggregation: On error resume next,");
				e.printStackTrace(System.err);
			}else{
				throw e;
			}
		}
	}

}